// @Author EthanScriptOn
// @Desc
package connect

import (
	"fmt"
	"gitee.com/fatzeng/srf_switch_basic_components/util"
	"gitee.com/fatzeng/srf_switch_expansion_package/detect"
	"github.com/confluentinc/confluent-kafka-go/kafka"
)

var KafkaProducer *util.OnceInitializer

var KafkaProducerChecker = func() (isInvalid bool) {
	producer, err := generateKafkaProducer()
	if err != nil {
		return true
	}
	if _, err = producer.GetMetadata(nil, false, 1000); err != nil {
		producer.Close()
		return true
	}
	return false
}

var KafkaProducerRecover = func() error {
	KafkaProducer.Reset()
	_, err := KafkaProducer.Get()
	return err
}

var KafkaProducerMonitor = detect.GenerateResourceMonitor("kafka_producer", true, KafkaProducerChecker, KafkaProducerRecover, detect.MonitorFunctionInstance)

func GetKafkaSyncProducer() (producer *kafka.Producer, err error) {
	defer detect.BackendHealthChecker.AddResource(KafkaProducerMonitor)
	return generateKafkaProducer()
}

func generateKafkaProducer() (producer *kafka.Producer, err error) {
	if KafkaProducer == nil {
		return nil, fmt.Errorf("the KafkaProducer initialization required")
	}
	instanceInter, err := KafkaProducer.Get()
	if err != nil {
		return nil, err
	}
	instance, ok := instanceInter.(*kafka.Producer)
	if !ok {
		return nil, fmt.Errorf("the instance is not of type *kafka.Producer")
	}
	return instance, nil
}

var KafkaConsumer *util.OnceInitializer

var KafkaConsumerChecker = func() (isInvalid bool) {
	producer, err := generateKafkaProducer()
	if err != nil {
		return true
	}
	if _, err = producer.GetMetadata(nil, false, 1000); err != nil {
		producer.Close()
		return true
	}
	return false
}

var KafkaConsumerRecover = func() error {
	KafkaConsumer.Reset()
	_, err := KafkaConsumer.Get()
	return err
}

var KafkaConsumerMonitor = detect.GenerateResourceMonitor("kafka_consumer", true, KafkaConsumerChecker, KafkaConsumerRecover, detect.MonitorFunctionInstance)

func GetKafkaConsume() (consumer *kafka.Consumer, err error) {
	defer detect.BackendHealthChecker.AddResource(KafkaConsumerMonitor)
	return generateKafkaConsumer()
}

func generateKafkaConsumer() (consumer *kafka.Consumer, err error) {
	if KafkaConsumer == nil {
		return nil, fmt.Errorf("the KafkaConsumer initialization required")
	}
	instanceInter, err := KafkaConsumer.Get()
	if err != nil {
		return nil, err
	}
	instance, ok := instanceInter.(*kafka.Consumer)
	if !ok {
		return nil, fmt.Errorf("the instance is not of type kafka.Consumer")
	}
	return instance, nil
}
